#include "include/ODUpdater.h"
#include "include/exception/UpdaterException.h"
#include "include/utils/VersionUtils.hpp"
#include "include/exception/UnConnectServerException.h"
#include "include/config/proceessing/UpdateTaskRunnable.h"
#include "include/config/handler/DefaultUpdaterHandler.h"
#include "include/config/proceessing/search/IntelligentSearchProcessor.h"
#include "include/config/server_api/DefaultUpdaterAPI.h"
#include "include/config/proceessing/file/DefaultFileProcessor.h"
#include "include/config/proceessing/record/DefaultUpdateRecorder.h"
#include "include/config/transmit/BasisStreamTransmit.h"
#include <iostream>
#include <utility>
#include <nlohmann/json.hpp>
#include "include/config/server_api/result_obj/ReqValue.h"

const string &ODUpdater::getIp() const {
    return ip;
}

int ODUpdater::getUnpublishKey() const {
    return unpublishKey;
}

int64_t ODUpdater::getLabelId() const {
    return labelId;
}

const string &ODUpdater::getReqProtocol() const {
    return reqProtocol;
}

IUpdateHandler *ODUpdater::getUpdateHandler() const {
    return updateHandler;
}

IResSearchProcessor *ODUpdater::getResSearchProcessor() const {
    return resSearchProcessor;
}

IUpdateRecorder *ODUpdater::getUpdateRecorder() const {
    return updateRecorder;
}

const string &ODUpdater::getRootDirectory() const {
    return rootDirectory;
}

list<UpdateError> ODUpdater::getErrorList() const {
    return errorList;
}

ODUpdater::~ODUpdater() {
    delete updateHandler;
    delete resSearchProcessor;
    delete updaterServer;
    delete fileProcessor;
    delete updateRecorder;
    delete logger;
    delete streamTransmit;
    delete consumerThreadPool;
}

BasisLogger *ODUpdater::getLogger() const {
    return logger;
}

IStreamTransmit *ODUpdater::getStreamTransmit() const {
    return streamTransmit;
}

IUpdaterServer *ODUpdater::getUpdaterServer() const {
    return updaterServer;
}

IFileProcessor *ODUpdater::getFileProcessor() const {
    return fileProcessor;
}

bool ODUpdater::startUpdate() noexcept(false){
    try{
        return update();
    }catch (exception &e){
        isStart = false;
        logger->error(e.what());
        throw e;
    }
}

bool ODUpdater::update() {

    if (isStart){
        throw UpdaterException("更新程序已经在运行中");
    }

    logger->info("更新初始化...");

    endUpdateCount = 0;
    errorList.clear();
    isStart = true;
    stop = false;

    logger->info("加载本地当前资源版本");
    UpdateRecord updateRecord = updateRecorder->getUpdatedVersion(*this);

    logger->debug("当前版本：" + version::intToVersion(updateRecord.getVersion()) + " | 未更新完成版本：" + version::intToVersion(updateRecord.getNotCompletedVersion()));

    //从小到大排序
    //Map<版本号,资源列表>
    map<int,list<Res>,less<>> needUpdate{};

    //通过资源检索处理器获取需要更新的资源
    //将资源放入 needUpdate 中
    logger->info("开始检索需要更新的资源");
    resSearchProcessor->getUpdateRes(needUpdate,*this,updateRecord);

    if (needUpdate.empty()){
        isStart = false;
        return false;
    }

    if (logger->isDebug()){
        for (const auto &item: needUpdate){
            logger->debug("================[v" + version::intToVersion(item.first) + "]================");
            for (const auto &res: item.second){
                logger->debug("资源-->id：" + to_string(res.getId()) + " | 是否为删除类型：" + (res.isDelete() ? "是" : "否") + " | 路径：" + res.getResPath());
            }
        }
    }

    logger->debug("获取最大并发下载数量");
    ReqValue<ValueData> reqConcurrencyCount = updaterServer->getConcurrencyCount(*this);
    if (reqConcurrencyCount.getServerApiState() != SUCCESS){
        throw UnConnectServerException("无法连接至服务器");
    }

    int concurrencyCount = std::stoi(reqConcurrencyCount.getData().getValue());

    logger->debug("并发下载数量-->" + to_string(concurrencyCount));

    delete consumerThreadPool;
    consumerThreadPool = new ThreadPool(concurrencyCount, concurrencyCount);

    //发布更新开始前事件
    UpdateStartEvent updateStartEvent(&needUpdate);
    updateHandler->updateStart(updateStartEvent,this);
    if (updateStartEvent.isCancel()){
        isStart = false;
        logger->info("取消更新");
        return false;
    }

    if(updateProducer.joinable()){
        updateProducer.join();
    }

    //启动生产者线程，遍历创建更新任务
    updateProducer = thread([this,needUpdate]{
        logger->info("=========================初始化完毕！更新启动！=========================");
        logger->debug("生产者线程启动，正在创建更新任务...");

        //最后一次更新的版本
        int nowVersion = 0;

        for (auto &item : needUpdate){
            list<Res> resList = item.second;
            nowVersion = item.first;
            updateHandler->updateStartVersion(nowVersion);
            if(logger->isEnabled()){
                logger->info("正在更新版本------------>[v" + version::intToVersion(nowVersion) + "]<------------");
            }

            //记录当前版本更新进度
            size_t endIndex = endUpdateCount + resList.size();

            for (auto &res : resList) {
                PreUpdateEvent preUpdateEvent(&res,nowVersion);
                updateHandler->preUpdate(preUpdateEvent);
                if (preUpdateEvent.isCancel()){
                    logger->info("取消资源更新-->id：" + to_string(res.getId()) + " | 是否为删除类型：" + (res.isDelete() ? "是" : "否") + " | 路径：" + res.getResPath());
                    continue;
                }

                //并发更新，线程池满时阻塞，优化内存
                try{
                    this->consumerThreadPool->execute([this,res,nowVersion]{
                        UpdateTaskRunnable::updateTask(*this,&res,nowVersion);
                    });
                }catch (exception &e){
                    UpdateError updateError(&res,e);
                    this->errorList.push_back(updateError);
                }

                //出错后跳出循环
                if (!this->errorList.empty() || this->stop){
                    break;
                }

            }

            if (this->errorList.empty() && !this->stop && endUpdateCount < endIndex){

                //等待当前版本所有任务完成
                {
                    unique_lock<std::mutex> lock(this->produceMutex);
                    if (logger->isEnabled()){
                        logger->debug("等待当前版本所有更新结束... " + to_string(endUpdateCount) + "/" + to_string(endIndex));
                    }
                    this->produceCondition.wait(lock,[this,&endIndex]{return this->endUpdateCount >= endIndex || this->stop;});
                }
            }

            //出错后跳出循环
            if (!this->errorList.empty() || this->stop){
                this->consumerThreadPool->shutdownAndDiscard();
                break;
            }

            this->logger->info("版本更新完成------------>[v" + version::intToVersion(nowVersion) + "]<------------");
            updateHandler->updateDoneVersion(nowVersion);
        }

        this->consumerThreadPool->shutdown();
        this->consumerThreadPool->await();

        this->isStart = false;
        if(this->stop){
            this->stop = true;
        }

        logger->info("=====================================================================");
        logger->info("更新完成！异常数量->" + to_string(errorList.size()));

        UpdateCompleteEvent updateCompleteEvent(errorList.empty() ? 0 : nowVersion,errorList,endUpdateCount);
        updateHandler->allUpdateComplete(updateCompleteEvent);
    });

    return true;
}

ODUpdater::ODUpdater(string ip, int unpublishKey, int64_t labelId, IUpdateHandler *updateHandler,
                     IResSearchProcessor *resSearchProcessor, IUpdaterServer *updaterServer,
                     IFileProcessor *fileProcessor, IUpdateRecorder *updateRecorder,
                     BasisLogger *logger, IStreamTransmit *streamTransmit, string rootDirectory) : ip(std::move(ip)),
                                                                                                          unpublishKey(
                                                                                                                  unpublishKey),
                                                                                                          labelId(labelId),
                                                                                                          updateHandler(
                                                                                                                  updateHandler),
                                                                                                          resSearchProcessor(
                                                                                                                  resSearchProcessor),
                                                                                                          updaterServer(
                                                                                                                  updaterServer),
                                                                                                          fileProcessor(
                                                                                                                  fileProcessor),
                                                                                                          updateRecorder(
                                                                                                                  updateRecorder),
                                                                                                          logger(logger),
                                                                                                          streamTransmit(
                                                                                                                  streamTransmit),
                                                                                                          rootDirectory(std::move(
                                                                                                                  rootDirectory)) {}

void ODUpdater::awaitUpdate() {
    if (updateProducer.joinable()){
        updateProducer.join();
    }
}

void ODUpdater::shutdown() {
    stop = true;
    if (consumerThreadPool != nullptr){
        consumerThreadPool->shutdownAndDiscard();
    }
    logger->debug("更新终止！");
}

void ODUpdater::addEndTask() {
    endUpdateCount++;
    produceCondition.notify_one();
    taskCondition.notify_one();
}

void ODUpdater::reducePoolSize() {
    {
        unique_lock<std::mutex> lock(this->produceMutex);
        if(consumerThreadPool->getCoreSize() > 1){
            consumerThreadPool->setCoreSize(consumerThreadPool->getCoreSize() - 1);
        }
    }
}

void ODUpdater::awaitOneTaskDone() {
    unique_lock<std::mutex> lock(this->taskMutex);
    taskCondition.wait(lock);
}

void ODUpdater::addUpdateError(const UpdateError &error) {
    errorList.push_back(error);
}

ODUpdater ODUpdater::createUpdater(const string &ip, int64_t labelId, int unpublishKey,const string &rootDir) {
    return {ip, unpublishKey, labelId,
                     new DefaultUpdaterHandler,new IntelligentSearchProcessor,
                     new DefaultUpdaterAPI,new DefaultFileProcessor,new DefaultUpdateRecorder,
                     new BasisLogger,new BasisStreamTransmit,rootDir};
}

void ODUpdater::setIp(const string &_ip) {
    tryModifyConfig();
    ODUpdater::ip = _ip;
}

void ODUpdater::setUnpublishKey(int _unpublishKey) {
    tryModifyConfig();
    ODUpdater::unpublishKey = _unpublishKey;
}

void ODUpdater::setLabelId(int64_t _labelId) {
    tryModifyConfig();
    ODUpdater::labelId = _labelId;
}

void ODUpdater::setReqProtocol(const string &_reqProtocol) {
    tryModifyConfig();
    ODUpdater::reqProtocol = _reqProtocol;
}

void ODUpdater::setUpdateHandler(IUpdateHandler *_updateHandler) {
    tryModifyConfig();
    delete updateHandler;
    ODUpdater::updateHandler = _updateHandler;
}

void ODUpdater::setResSearchProcessor(IResSearchProcessor *_resSearchProcessor) {
    tryModifyConfig();
    delete resSearchProcessor;
    ODUpdater::resSearchProcessor = _resSearchProcessor;
}

void ODUpdater::setUpdaterServer(IUpdaterServer *_updaterServer) {
    tryModifyConfig();
    delete updaterServer;
    ODUpdater::updaterServer = _updaterServer;
}

void ODUpdater::setFileProcessor(IFileProcessor *_fileProcessor) {
    tryModifyConfig();
    delete fileProcessor;
    ODUpdater::fileProcessor = _fileProcessor;
}

void ODUpdater::setUpdateRecorder(IUpdateRecorder *_updateRecorder) {
    tryModifyConfig();
    delete updateRecorder;
    ODUpdater::updateRecorder = _updateRecorder;
}

void ODUpdater::setRootDirectory(const string &_rootDirectory) {
    tryModifyConfig();
    ODUpdater::rootDirectory = _rootDirectory;
}

void ODUpdater::setLogger(BasisLogger *_logger) {
    tryModifyConfig();
    delete logger;
    ODUpdater::logger = _logger;
}

void ODUpdater::setStreamTransmit(IStreamTransmit *_streamTransmit) {
    tryModifyConfig();
    delete streamTransmit;
    ODUpdater::streamTransmit = _streamTransmit;
}

void ODUpdater::tryModifyConfig() const{
    if (isStart){
        throw UpdaterException("更新器运行期间不可修改配置！");
    }
}

ODUpdater ODUpdater::createUpdater(const string &ip, int64_t labelId) {
    return createUpdater(ip,labelId,0,"./");
}

ODUpdater ODUpdater::createUpdater(const string &ip, int64_t labelId, const string &rootDirectory) {
    return createUpdater(ip,labelId,0,rootDirectory);
}

bool ODUpdater::isStartUpdate() const {
    return this->isStart;
}
